[AWS IoT] RaspberryPiからMQTTで画像をS3にアップロードしてみました

MQTT　画像も送れる　すぐれも

#AWS

#MQTT

#AWS IoT

SIN

2020.08.05

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

1 はじめに

CX事業本部の平内（SIN）です。

今回は、AWS IoTのメッセージブローカーを経由して、MQTTでRaspberryPiからS3に画像を送信する仕組み試してみました。

この仕組みでは、デバイス側に、S3アクセスのためのクレデンシャル情報は必要ありません。

また、送信先となるS3のバケット名やファイル（プレフィックス）名は、デバイス側で指定する仕組みとしてみました。

想定する利用場面以下のとおりです。

RaspberryPiに接続したカメラで動画を撮影
動画をフレーム単位で送信
IoT Coreでは、ルールでLambdaを起動しS3に蓄積
S3の画像ファイルを動画ファイルにマージする

なお、赤色の点線内は、今回、未実装です。RaspberryPiからは、予め用意された画像ファイルを読み込んで送信しています。

2 画像ファイル

送信に使用した画像ファイルです。

先日行ってきた、うちの近くにある犬のだがし屋さん(旧：犬の焼きいも屋さん)の写真です。

640 * 480 で 97KByte となっています。

3 ルール

定義されたルールです。

指定したトピック（ここでは、topic_1となっています）を受信した場合に、Lambda(BinaryUploadTest)に送っています。

4 Lambda

ルールから起動されるLambdaのコードです。

受信したPayloadから、S3のバケット名、キー名及び、エンコードされた画像を取得し、S3にUploadしています。

S3へのUpload権限は、別途、このLambda関数に付与されています。

import json
import os
import boto3
from datetime import datetime
from io import BytesIO
from base64 import b64decode

def lambda_handler(event, context):
    print("event",event)

    img = event["img"]
    bucket = event["bucket"]
    key = event["key"]

    # デコード base64 => binary
    img_binary = b64decode(img)

    s3 = boto3.resource('s3')
    obj = s3.Object(bucket, key)
    obj.put(Body=BytesIO(img_binary),ContentType="image/jpg")

    return 0

5 クライアント

RaspberryPi上で動作しているコードは、以下のとおりです。

payloadには、バケット名、キー名、及び、Base64でエンコードされた画像が含まれています。

コード側で、バケット名及び、キー名を指定していますが、S3への権限は、Lambdaに付与が必要です。

index.py

from mqtt import Mqtt
import time
import datetime
import json
import base64

endPoint = "xxxxxxxxxxx-ats.iot.ap-northeast-1.amazonaws.com"
port = 8883
rootCA = "./cert/root-CA.crt"
privateKey = "./cert/private.key"
certificate = "./cert/cert.pem"

clientId = "client_id"
topic = "topic_1"

bucket = "binary-upload-test-2020-08-05"

def main():
    mqtt = Mqtt(endPoint, port, rootCA, privateKey, certificate, clientId)

    # 送信する画像ファイルを読み込む
    img = open("./photo.jpg", 'rb').read()
    
    while True:
        # 現在時間を使用して、ファイル名を生成
        dt = datetime.datetime.fromtimestamp(time.time())
        key = "{}{}.jpg".format(dt.strftime('%H%M'), dt.strftime('%S'))
        
        # エンコード binary => base64
        uploadImg = base64.b64encode(img).decode("utf-8")
        
        # 送信
        mqtt.publish(topic, payload=json.dumps({
            "img": uploadImg,
            "bucket": bucket,
            "key": key
        }))
        
        print("publish: {}".format(key))
        time.sleep(5)

main()

mqtt.py

# MQTT pub/subの基本クラス

from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient
from log import logger

class Mqtt():
    def __init__(self, endPoint, port, rootCA, privateKey, certificate, clientId):
        logger.debug("Mqtt.__init__()")
        self.__client = AWSIoTMQTTClient(clientId)
        self.__client.configureEndpoint(endPoint, port)
        self.__client.configureCredentials(rootCA, privateKey, certificate)
        self.__client.configureAutoReconnectBackoffTime(1, 32, 20)
        self.__client.configureOfflinePublishQueueing(-1)
        self.__client.configureDrainingFrequency(2)
        self.__client.configureConnectDisconnectTimeout(10)
        self.__client.configureMQTTOperationTimeout(5)
        self.__client.connect()
    
    def subscribe(self, topic, callback):
        logger.debug("Mqtt.subscribe()")
        self.__client.subscribe(topic, 1, callback)

    def publish(self, topic, payload):
        result = self.__client.publish(topic, payload, 1)

6 実行

先のコードを実行すると、指定したバケットにファイルが保存されることが確認できます。

$ python3 index.py
2020-08-05 09:15:31,764 - Sample - DEBUG - Mqtt.__init__()
publish: 091532.jpg
publish: 091537.jpg
publish: 091542.jpg
publish: 091547.jpg